#include "../lib/servermap.hpp"
#include "../lib/protocol.hpp"
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <netdb.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <poll.h>
#include <boost/algorithm/string.hpp>

#include <iostream>
using namespace std;

using namespace pipc;

ServerMap::ServerMap(std::string ip, int port,std::string id)
{
    port_ = port;
    ip_ = ip;
    id_ = id;
    sockfd_ = -1;
    running_ = false;
}

ServerMap::~ServerMap()
{
    if(running_)
        stop();

}

bool ServerMap::setValue(std::string key, Record value)
{
    boost::unique_lock<boost::shared_mutex > lock(serializer_);
    if(settings_.find(key) != settings_.end())
    {
        if(marks_[key] == false)
            return false;
    }
    settings_[key] = value;
    marks_[key] = false;
    ack_[key] = true;
    if(running_)
    {
        CmdHeader header;
        header.cmd = HEADER_SET_COMMAND;
        header.tv_sec = value.effective.tv_sec;
        header.tv_usec  = value.effective.tv_usec;

        std::string data = key + PIPC_PROTOCOL_DELIMITER + value.data;
        header.payloadLen = data.length();
        ::write(sockfd_, &header, sizeof(header));
        ::write(sockfd_, data.c_str(), data.length());
    }
    return true;
}

Record ServerMap::getValue(std::string key)
{
    Record ret;
    timeval tv;
    boost::shared_lock<boost::shared_mutex> lock(serializer_);
    if(settings_.find(key) == settings_.end())
    {
        gettimeofday(&(ret.effective), NULL);
        return ret;;
    }

    if(running_ )
    {
        if(ack_[key] == false)
        {
            gettimeofday(&(tv), NULL);
            CmdHeader header;
            header.cmd = HEADER_RECIEVED_COMMAND;
            header.tv_sec = tv.tv_sec;
            header.tv_usec  = tv.tv_usec;
            std::string data = key;
            header.payloadLen = data.length();
            ::write(sockfd_, &header, sizeof(header));
            ::write(sockfd_, data.c_str(), data.length());
            ack_[key] = true;
        }
    }
    return settings_.at(key);
}

void ServerMap::clear()
{
    boost::unique_lock< boost::shared_mutex > lock(serializer_);
    settings_.clear();
}

bool ServerMap::start()
{
    if(running_)
        return false;
    boost::unique_lock<boost::shared_mutex > lock(serializer_);
    listener_ = new boost::thread(&ServerMap::start_, this);
    return true;
}

bool ServerMap::stop()
{
    boost::unique_lock<boost::shared_mutex > lock(serializer_);
    if(!running_)
        return false;
    listener_->interrupt();
    listener_->join();
    delete listener_;
    return true;
}

bool ServerMap::isRunning()
{
    boost::unique_lock<boost::shared_mutex > lock(serializer_);
    return running_;
}

void ServerMap::markRecieved_(std::string key)
{
    boost::unique_lock<boost::shared_mutex > lock(serializer_);
    marks_[key] = true;
}

bool ServerMap::isSynced(std::string key)
{
    boost::shared_lock<boost::shared_mutex> lock(serializer_);
    return marks_[key];
}

bool ServerMap::subscibe(std::string key, bool reader)
{
    boost::shared_lock<boost::shared_mutex> lock(serializer_);

    CmdHeader header;
    if(reader)
        header.cmd = HEADER_SUBSCRIBE_READER;
    else
        header.cmd = HEADER_SUBSCRIBE_WRITER;
    header.payloadLen = key.length();
    struct timeval tv;
    gettimeofday(&(tv), NULL);
    header.tv_sec = tv.tv_sec;
    header.tv_usec = tv.tv_usec;
    if(!running_)
        return false;
    ::write(sockfd_, &header, sizeof(header));
    ::write(sockfd_, key.c_str(), key.length());
    return true;
}

void ServerMap::start_()
{
    try
    {
        struct sockaddr_in serv_addr;
        struct hostent *server = gethostbyname(ip_.c_str());
        struct pollfd fds;
        struct timeval tv;
        {
            boost::unique_lock<boost::shared_mutex > lock(serializer_);
            running_ = true;
            if(server == NULL)
            {
                goto OUT;
            }
            sockfd_ = socket(AF_INET, SOCK_STREAM, 0);
            if (sockfd_ < 0)
            {
                goto OUT;
            }
            bzero((char *) &serv_addr, sizeof(serv_addr));

            serv_addr.sin_family = AF_INET;
            bcopy((char *)server->h_addr, (char *)&serv_addr.sin_addr.s_addr, server->h_length);
            serv_addr.sin_port = htons(port_);
            if (connect(sockfd_,(struct sockaddr *) &serv_addr,sizeof(serv_addr)) != 0)
            {
                goto OUT;
            }
            gettimeofday(&(tv), NULL);
            CmdHeader header;
            header.cmd = HEADER_HANDSHAKE_COMMAND;
            header.tv_sec = tv.tv_sec;
            header.tv_usec = tv.tv_usec;
            header.payloadLen = id_.length();
            ::write(sockfd_, &header, sizeof(header));
            ::write(sockfd_, id_.c_str(), id_.length());

        }
        fds.fd = sockfd_;
        fds.events = POLLIN;

        while(true)
        {
            boost::this_thread::interruption_point();
            int retval;
            retval = poll(&fds,1,500);
            if(retval < 0)
                return;
            if(retval == 0)
                continue;
            CmdHeader header;
            retval = ::read(sockfd_, &header, sizeof(header));
            if(retval != sizeof(header))
                goto OUT;
            std::vector<std::string> args;
            if(header.payloadLen >0)
            {
                char buff[header.payloadLen + 1];
                retval = ::read(sockfd_, &buff, header.payloadLen);
                if(retval != header.payloadLen)
                    goto OUT;
                buff[header.payloadLen] = '\0';
                std::string stringBuff(buff);
                boost::split(args,stringBuff,boost::is_any_of(PIPC_PROTOCOL_DELIMITER));
            }
            switch(header.cmd)
            {
            case HEADER_SET_NO_ACK_COMMAND:
            case HEADER_SET_COMMAND:
            {

                if(args.size() <2)
                    break;
                {
                    std::string key = args[0];
                    boost::unique_lock<boost::shared_mutex > lock(serializer_);
                    Record rec;
                    rec.data = std::string(args[1]);
                    rec.effective.tv_sec = header.tv_sec;
                    rec.effective.tv_usec = header.tv_usec;
                    settings_[key] = rec;
                    if(header.cmd == HEADER_SET_COMMAND)
                        ack_[key] = false;
                    else
                        ack_[key] = true;
                    marks_[key] = true;
                }
                break;
            }
            case HEADER_RECIEVED_COMMAND:
                if(args.size() < 1)
                    break;
                else
                {
                    markRecieved_(args[0]);
                }

            }
        }
    }
    catch(boost::thread_interrupted e){}
OUT:
    if(sockfd_ > 0)
    {
        running_ = false;
        close(sockfd_);
    }
    return;
}
